package com.example.monit.service;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch._types.query_dsl.RangeQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.TermQuery;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.sql.QueryResponse;
import co.elastic.clients.json.JsonData;
import com.example.monit.bean.InterfaceIndicator;
import com.example.monit.bean.ResponseRust;
import com.example.monit.dto.QueryInterfaceIndicatorDto;
import com.example.monit.dto.TotalDto;
import com.example.monit.enums.IndexName;
import com.example.monit.utils.TimeUtils;
import com.example.monit.vo.InterfaceIndicatorTotalVo;
import com.example.monit.vo.InterfaceIndicatorVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class InterfaceIndicatorServiceImpl extends BasEsService implements InterfaceIndicatorService {

    @Autowired
    ElasticsearchClient client;

    @Override
    public ResponseRust upLoadInterfaceIndicator(List<InterfaceIndicator> interfaceIndicator) throws IOException {
//        List<BulkOperation> operationList = interfaceIndicator.stream().map(e ->
//                        BulkOperation.of(builder ->
//                                builder.index(builder1 -> builder1.document(e).index(IndexName.InterfaceIndicator))))
//                .collect(Collectors.toList());
//        BulkResponse response = client.bulk(builder -> builder.operations(operationList));
//        if (response.errors()) {
//            log.error("上报数据异常错误");
//        }
        return ResponseRust.success_creat();
    }

    @Override
    public ResponseRust queryInterfaceIndicator(InterfaceIndicatorVo interfaceIndicatorVo) throws IOException {
        String sqlString = " SELECT url,count(url),userID,pageUrl,sum(duration)\n" +
                "            FROM \"interface_indicator\"\n" +
                "            where appId='%s' and mainType=%d and subType=%d and startTime between %d and %d\n" +
                "            group by url ,userID,pageUrl\n" +
                "            order by count(url) desc";

        sqlString = String.format(sqlString,
                interfaceIndicatorVo.getApp_id(),
                interfaceIndicatorVo.getMain_type(),
                interfaceIndicatorVo.getSub_type(),
                interfaceIndicatorVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                interfaceIndicatorVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli());//todo 时区问题
        String finalSqlString = sqlString;
        QueryResponse query = client.sql().query(e -> e.query(finalSqlString));

        List<List<JsonData>> list = query.rows();

        HashMap<String, QueryInterfaceIndicatorDto> map = new HashMap<>();
        for (List<JsonData> jsonData : list) {
            String url = jsonData.get(0).to(String.class);
            int count = jsonData.get(1).to(Integer.class);
            String userID = jsonData.get(2).to(String.class);
            String pageUrl = jsonData.get(3).to(String.class);
            BigDecimal sumValue = jsonData.get(4).to(BigDecimal.class);

            QueryInterfaceIndicatorDto build = QueryInterfaceIndicatorDto.builder()
                    .url(url)
                    .count(count)
                    .userCount(1)
                    .pageCount(1)
                    .userList(Arrays.asList(userID))
                    .pageList(Arrays.asList(pageUrl))
                    .sumAverage(sumValue)
                    .build();

            String key = url;
            QueryInterfaceIndicatorDto queryDto = map.get(key);
            if (ObjectUtils.isEmpty(queryDto)) {
                map.put(pageUrl, build);
            } else {
                if (!queryDto.getUserList().contains(userID)) {
                    queryDto.setUserCount(queryDto.getUserCount() + 1);
                    queryDto.getUserList().add(userID);
                }
                if (!queryDto.getPageList().contains(pageUrl)) {
                    queryDto.setPageCount(queryDto.getPageCount() + 1);
                    queryDto.getPageList().add(pageUrl);
                }
                queryDto.setCount(queryDto.getCount() + build.getCount());
            }
        }

        Collection<QueryInterfaceIndicatorDto> values = map.values();
        for (QueryInterfaceIndicatorDto queryDto : values) {
            if (!queryDto.getSumAverage().equals(BigDecimal.ZERO)) {
                queryDto.setAverage(queryDto.getSumAverage().divide(new BigDecimal(queryDto.getCount()), 2, RoundingMode.HALF_UP));
            }
        }

        if (!ObjectUtils.isEmpty(interfaceIndicatorVo.getSize())) {
            values = values.stream().limit(interfaceIndicatorVo.getSize()).collect(Collectors.toList());
        }

        return ResponseRust.success(values);
    }

    @Override
    public ResponseRust totalInterfaceIndicator(InterfaceIndicatorTotalVo interfaceIndicatorTotalVo) throws IOException {
        SearchRequest searchRequest = SearchRequest.of(s ->
                s.index(IndexName.InterfaceIndicator)
                        .query(getQuery(interfaceIndicatorTotalVo))
                        .aggregations("count", getAggregation(interfaceIndicatorTotalVo))
                        .size(0));

        SearchResponse<InterfaceIndicator> response = client.search(searchRequest, InterfaceIndicator.class);
        Map<String, Aggregate> aggregations = response.aggregations();
        Aggregate count = aggregations.get("count");
        RangeAggregate range = count.range();

        LinkedList<List<TotalDto>> list = new LinkedList<>();
        // 5个范围
        for (RangeBucket bucket : range.buckets().array()) {
            Buckets<HistogramBucket> buckets = bucket.aggregations().get("list").histogram().buckets();
            list.add(totalData(buckets, interfaceIndicatorTotalVo));
        }
        return ResponseRust.success(list);
    }

    private Aggregation getAggregation(InterfaceIndicatorTotalVo interfaceIndicatorTotalVo) {
        List<AggregationRange> ranges = new LinkedList<>();
        ranges.add(AggregationRange.of(e -> e.from("0").to("1000")));
        ranges.add(AggregationRange.of(e -> e.from("1000").to("5000")));
        ranges.add(AggregationRange.of(e -> e.from("5000").to("10000")));
        ranges.add(AggregationRange.of(e -> e.from("10000").to("300000")));
        ranges.add(AggregationRange.of(e -> e.from("300000")));

        HashMap<String, Aggregation> aggregationHashMap = new HashMap<>(3);
        aggregationHashMap.put("avg", Aggregation.of(avg -> avg.avg(e -> e.field("duration"))));
        aggregationHashMap.put("userCount", Aggregation.of(userCount -> userCount.cardinality(e -> e.field("userID"))));
        aggregationHashMap.put("pageCount", Aggregation.of(pageCount -> pageCount.cardinality(e -> e.field("pageUrl"))));


        if (!StringUtils.hasLength(interfaceIndicatorTotalVo.getGranularity())) {
            interfaceIndicatorTotalVo.setGranularity("1d");
        }
        int intValue = Integer.parseInt(interfaceIndicatorTotalVo.getGranularity().split("[smhdMy]")[0]);
        String type = interfaceIndicatorTotalVo.getGranularity().split(intValue + "")[1];
        ChronoUnit unit = TimeUtils.getType(type);

        return Aggregation.of(count -> count.range(RangeAggregation.of(e -> e.ranges(ranges).field("duration")))
                .aggregations("list", list -> list.histogram(time -> time.interval((double) unit.getDuration().toMillis() * intValue)
                                .minDocCount(0)
                                .field("startTime"))
                        .aggregations(aggregationHashMap)));
    }

    private Query getQuery(InterfaceIndicatorTotalVo querys) {
        LinkedList<Query> list = new LinkedList<>();
        if (StringUtils.hasLength(querys.getUrl())) {
            list.add(TermQuery.of(t->t.field("url").value(e->e.stringValue(querys.getUrl())))._toQuery());
        }
        list.add(TermQuery.of(t->t.field("appId").value(e->e.stringValue(querys.getApp_id())))._toQuery());
        list.add(TermQuery.of(t->t.field("mainType").value(e->e.longValue(querys.getMain_type())))._toQuery());
        list.add(TermQuery.of(t->t.field("subType").value(e->e.longValue(querys.getSub_type())))._toQuery());
        list.add(RangeQuery.of(r->r.gte(JsonData.of(querys.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()))
                .lte(JsonData.of(querys.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()))
                .field("startTime"))._toQuery());

        return  Query.of(query -> query.bool(bool->bool.must(list)));
    }
}
